pydaptivefiltering

134class AdaptiveFilter(ABC):
135    """Abstract base class for all adaptive filters.
136
137    Parameters
138    ----------
139    filter_order:
140        Order in the FIR sense (number of taps - 1). For non-FIR structures, it can be used
141        as a generic size indicator for base allocation.
142    w_init:
143        Initial coefficient vector. If None, initialized to zeros.
144
145    Notes
146    -----
147    - Subclasses should set `supports_complex = True` if they support complex-valued data.
148    - Subclasses are expected to call `_record_history()` every iteration (or use helper methods)
149      if they want coefficient trajectories.
150    """
151
152    supports_complex: bool = False
153
154    def __init__(self, filter_order: int, w_init: Optional[ArrayLike] = None) -> None:
155        self.filter_order: int = int(filter_order)
156        self._dtype = complex if self.supports_complex else float
157
158        self.regressor: np.ndarray = np.zeros(self.filter_order + 1, dtype=self._dtype)
159
160        if w_init is not None:
161            self.w: np.ndarray = np.asarray(w_init, dtype=self._dtype)
162        else:
163            self.w = np.zeros(self.filter_order + 1, dtype=self._dtype)
164
165        self.w_history: List[np.ndarray] = []
166        self._record_history()
167
168    def _record_history(self) -> None:
169        """Store a snapshot of current coefficients."""
170        self.w_history.append(np.asarray(self.w).copy())
171
172    def _final_coeffs(self, coefficients: Any) -> Any:
173        """Return last coefficients from a history container (list or 2D array)."""
174        if coefficients is None:
175            return None
176        if isinstance(coefficients, list) and len(coefficients) > 0:
177            return coefficients[-1]
178        try:
179            a = np.asarray(coefficients)
180            if a.ndim == 2:
181                return a[-1, :]
182        except Exception:
183            pass
184        return coefficients
185
186    def _pack_results(
187        self,
188        outputs: np.ndarray,
189        errors: np.ndarray,
190        runtime_s: float,
191        error_type: str = "a_priori",
192        extra: Optional[Dict[str, Any]] = None,
193    ) -> OptimizationResult:
194        """Centralized output packaging to standardize results."""
195        return OptimizationResult(
196            outputs=np.asarray(outputs),
197            errors=np.asarray(errors),
198            coefficients=np.asarray(self.w_history),
199            algorithm=self.__class__.__name__,
200            runtime_ms=float(runtime_s) * 1000.0,
201            error_type=str(error_type),
202            extra=extra,
203        )
204
205    def filter_signal(self, input_signal: ArrayLike) -> np.ndarray:
206        """Filter an input signal using current coefficients.
207
208        Default implementation assumes an FIR structure with taps `self.w` and
209        regressor convention:
210            x_k = [x[k], x[k-1], ..., x[k-m]]
211        and output:
212            y[k] = w^H x_k   (Hermitian for complex)
213        """
214        x = np.asarray(input_signal, dtype=self._dtype)
215        n_samples = x.size
216        y = np.zeros(n_samples, dtype=self._dtype)
217
218        x_padded = np.zeros(n_samples + self.filter_order, dtype=self._dtype)
219        x_padded[self.filter_order:] = x
220
221        for k in range(n_samples):
222            x_k = x_padded[k : k + self.filter_order + 1][::-1]
223            y[k] = np.dot(self.w.conj(), x_k)
224
225        return y
226
227    @classmethod
228    def default_test_init_kwargs(cls, order: int) -> dict:
229        """Override in subclasses to provide init kwargs for standardized tests."""
230        return {}
231
232    @abstractmethod
233    def optimize(
234        self,
235        input_signal: ArrayLike,
236        desired_signal: ArrayLike,
237        **kwargs: Any,
238    ) -> Any:
239        """Run the adaptation procedure.
240
241        Subclasses should return either:
242        - OptimizationResult (recommended), or
243        - dict-like with standardized keys, if you are migrating older code.
244        """
245        raise NotImplementedError
246
247    def reset_filter(self, w_new: Optional[ArrayLike] = None) -> None:
248        """Reset coefficients and history."""
249        if w_new is not None:
250            self.w = np.asarray(w_new, dtype=self._dtype)
251        else:
252            self.w = np.zeros(self.filter_order + 1, dtype=self._dtype)
253        self.w_history = []
254        self._record_history()

 28class LMS(AdaptiveFilter):
 29    """
 30    Complex Least-Mean Squares (LMS) adaptive filter.
 31
 32    Standard complex LMS algorithm for adaptive FIR filtering, following Diniz
 33    (Alg. 3.2). The method performs a stochastic-gradient update using the
 34    instantaneous a priori error.
 35
 36    Parameters
 37    ----------
 38    filter_order : int
 39        Adaptive FIR filter order ``M``. The number of coefficients is ``M + 1``.
 40    step_size : float, optional
 41        Adaptation step size ``mu``. Default is 1e-2.
 42    w_init : array_like of complex, optional
 43        Initial coefficient vector ``w(0)`` with shape ``(M + 1,)``. If None,
 44        initializes with zeros.
 45
 46    Notes
 47    -----
 48    At iteration ``k``, form the regressor vector (newest sample first):
 49
 50    .. math::
 51        x_k = [x[k], x[k-1], \\ldots, x[k-M]]^T \\in \\mathbb{C}^{M+1}.
 52
 53    The a priori output and error are
 54
 55    .. math::
 56        y[k] = w^H[k] x_k, \\qquad e[k] = d[k] - y[k],
 57
 58    and the LMS update is
 59
 60    .. math::
 61        w[k+1] = w[k] + \\mu\\, e^*[k] \\, x_k.
 62
 63    This implementation:
 64        - uses complex arithmetic (``supports_complex=True``),
 65        - returns the a priori error ``e[k]``,
 66        - records coefficient history via the base class.
 67
 68    References
 69    ----------
 70    .. [1] P. S. R. Diniz, *Adaptive Filtering: Algorithms and Practical
 71       Implementation*, 5th ed., Algorithm 3.2.
 72    """
 73
 74    supports_complex: bool = True
 75
 76    step_size: float
 77
 78    def __init__(
 79        self,
 80        filter_order: int,
 81        step_size: float = 1e-2,
 82        w_init: Optional[ArrayLike] = None,
 83    ) -> None:
 84        super().__init__(filter_order=int(filter_order), w_init=w_init)
 85        self.step_size = float(step_size)
 86
 87    @validate_input
 88    def optimize(
 89        self,
 90        input_signal: np.ndarray,
 91        desired_signal: np.ndarray,
 92        verbose: bool = False,
 93    ) -> OptimizationResult:
 94        """
 95        Executes the LMS adaptation loop over paired input/desired sequences.
 96
 97        Parameters
 98        ----------
 99        input_signal : array_like of complex
100            Input sequence ``x[k]`` with shape ``(N,)`` (will be flattened).
101        desired_signal : array_like of complex
102            Desired sequence ``d[k]`` with shape ``(N,)`` (will be flattened).
103        verbose : bool, optional
104            If True, prints the total runtime after completion.
105
106        Returns
107        -------
108        OptimizationResult
109            Result object with fields:
110            - outputs : ndarray of complex, shape ``(N,)``
111                Scalar output sequence, ``y[k] = w^H[k] x_k``.
112            - errors : ndarray of complex, shape ``(N,)``
113                Scalar a priori error sequence, ``e[k] = d[k] - y[k]``.
114            - coefficients : ndarray of complex
115                Coefficient history recorded by the base class.
116            - error_type : str
117                Set to ``"a_priori"``.
118        """
119        tic: float = perf_counter()
120
121        x: np.ndarray = np.asarray(input_signal, dtype=complex).ravel()
122        d: np.ndarray = np.asarray(desired_signal, dtype=complex).ravel()
123
124        n_samples: int = int(x.size)
125        m: int = int(self.filter_order)
126
127        outputs: np.ndarray = np.zeros(n_samples, dtype=complex)
128        errors: np.ndarray = np.zeros(n_samples, dtype=complex)
129
130        x_padded: np.ndarray = np.zeros(n_samples + m, dtype=complex)
131        x_padded[m:] = x
132
133        for k in range(n_samples):
134            x_k: np.ndarray = x_padded[k : k + m + 1][::-1]
135
136            y_k: complex = complex(np.vdot(self.w, x_k)) 
137            outputs[k] = y_k
138    
139            e_k: complex = d[k] - y_k
140            errors[k] = e_k
141
142            self.w = self.w + self.step_size * np.conj(e_k) * x_k
143
144            self._record_history()
145
146        runtime_s: float = perf_counter() - tic
147        if verbose:
148            print(f"[LMS] Completed in {runtime_s * 1000:.03f} ms")
149
150        return self._pack_results(
151            outputs=outputs,
152            errors=errors,
153            runtime_s=runtime_s,
154            error_type="a_priori",
155        )

 28class NLMS(AdaptiveFilter):
 29    """
 30    Complex Normalized Least-Mean Squares (NLMS) adaptive filter.
 31
 32    Normalized LMS algorithm for adaptive FIR filtering, following Diniz
 33    (Alg. 4.3). The method normalizes the step size by the instantaneous
 34    regressor energy to improve stability and reduce sensitivity to input
 35    scaling.
 36
 37    Parameters
 38    ----------
 39    filter_order : int
 40        Adaptive FIR filter order ``M``. The number of coefficients is ``M + 1``.
 41    step_size : float, optional
 42        Base adaptation step size ``mu``. Default is 1e-2.
 43    gamma : float, optional
 44        Regularization constant ``gamma`` used in the normalization denominator
 45        to avoid division by zero (or near-zero regressor energy). Default is 1e-6.
 46    w_init : array_like of complex, optional
 47        Initial coefficient vector ``w(0)`` with shape ``(M + 1,)``. If None,
 48        initializes with zeros.
 49
 50    Notes
 51    -----
 52    At iteration ``k``, form the regressor vector (newest sample first):
 53
 54    .. math::
 55        x_k = [x[k], x[k-1], \\ldots, x[k-M]]^T \\in \\mathbb{C}^{M+1}.
 56
 57    The a priori output and error are
 58
 59    .. math::
 60        y[k] = w^H[k] x_k, \\qquad e[k] = d[k] - y[k].
 61
 62    Define the instantaneous regressor energy
 63
 64    .. math::
 65        \\|x_k\\|^2 = x_k^H x_k,
 66
 67    and the normalized step size
 68
 69    .. math::
 70        \\mu_k = \\frac{\\mu}{\\|x_k\\|^2 + \\gamma}.
 71
 72    The NLMS update is then
 73
 74    .. math::
 75        w[k+1] = w[k] + \\mu_k\\, e^*[k] \\, x_k.
 76
 77    This implementation:
 78        - uses complex arithmetic (``supports_complex=True``),
 79        - returns the a priori error ``e[k]``,
 80        - records coefficient history via the base class.
 81
 82    References
 83    ----------
 84    .. [1] P. S. R. Diniz, *Adaptive Filtering: Algorithms and Practical
 85       Implementation*, 5th ed., Algorithm 4.3.
 86    """
 87
 88    supports_complex: bool = True
 89
 90    step_size: float
 91    gamma: float
 92
 93    def __init__(
 94        self,
 95        filter_order: int,
 96        step_size: float = 1e-2,
 97        gamma: float = 1e-6,
 98        w_init: Optional[ArrayLike] = None,
 99    ) -> None:
100        super().__init__(filter_order=int(filter_order), w_init=w_init)
101        self.step_size = float(step_size)
102        self.gamma = float(gamma)
103
104    @validate_input
105    def optimize(
106        self,
107        input_signal: np.ndarray,
108        desired_signal: np.ndarray,
109        verbose: bool = False,
110    ) -> OptimizationResult:
111        """
112        Executes the NLMS adaptation loop over paired input/desired sequences.
113
114        Parameters
115        ----------
116        input_signal : array_like of complex
117            Input sequence ``x[k]`` with shape ``(N,)`` (will be flattened).
118        desired_signal : array_like of complex
119            Desired sequence ``d[k]`` with shape ``(N,)`` (will be flattened).
120        verbose : bool, optional
121            If True, prints the total runtime after completion.
122
123        Returns
124        -------
125        OptimizationResult
126            Result object with fields:
127            - outputs : ndarray of complex, shape ``(N,)``
128                Scalar output sequence, ``y[k] = w^H[k] x_k``.
129            - errors : ndarray of complex, shape ``(N,)``
130                Scalar a priori error sequence, ``e[k] = d[k] - y[k]``.
131            - coefficients : ndarray of complex
132                Coefficient history recorded by the base class.
133            - error_type : str
134                Set to ``"a_priori"``.
135        """
136        tic: float = perf_counter()
137
138        x: np.ndarray = np.asarray(input_signal, dtype=complex).ravel()
139        d: np.ndarray = np.asarray(desired_signal, dtype=complex).ravel()
140
141        n_samples: int = int(x.size)
142        m: int = int(self.filter_order)
143
144        outputs: np.ndarray = np.zeros(n_samples, dtype=complex)
145        errors: np.ndarray = np.zeros(n_samples, dtype=complex)
146
147        x_padded: np.ndarray = np.zeros(n_samples + m, dtype=complex)
148        x_padded[m:] = x
149
150        for k in range(n_samples):
151            x_k: np.ndarray = x_padded[k : k + m + 1][::-1]
152
153            y_k: complex = complex(np.vdot(self.w, x_k))
154            outputs[k] = y_k
155
156            e_k: complex = d[k] - y_k
157            errors[k] = e_k
158
159            norm_xk: float = float(np.vdot(x_k, x_k).real)
160            mu_k: float = self.step_size / (norm_xk + self.gamma)
161
162            self.w = self.w + mu_k * np.conj(e_k) * x_k
163
164            self._record_history()
165
166        runtime_s: float = perf_counter() - tic
167        if verbose:
168            print(f"[NLMS] Completed in {runtime_s * 1000:.03f} ms")
169
170        return self._pack_results(
171            outputs=outputs,
172            errors=errors,
173            runtime_s=runtime_s,
174            error_type="a_priori",
175        )

 29class AffineProjection(AdaptiveFilter):
 30    """
 31    Complex Affine-Projection Algorithm (APA) adaptive filter.
 32
 33    Affine-projection LMS-type algorithm that reuses the last ``L+1`` regressor
 34    vectors to accelerate convergence relative to LMS/NLMS, following Diniz
 35    (Alg. 4.6). Per iteration, the method solves a small linear system of size
 36    ``(L+1) x (L+1)``.
 37
 38    Parameters
 39    ----------
 40    filter_order : int
 41        Adaptive FIR filter order ``M``. The number of coefficients is ``M + 1``.
 42    step_size : float, optional
 43        Adaptation step size (relaxation factor) ``mu``. Default is 1e-2.
 44    gamma : float, optional
 45        Diagonal loading (regularization) ``gamma`` applied to the projection
 46        correlation matrix for numerical stability. Default is 1e-6.
 47    L : int, optional
 48        Reuse factor (projection order). The algorithm uses ``L + 1`` most recent
 49        regressors. Default is 2.
 50    w_init : array_like of complex, optional
 51        Initial coefficient vector ``w(0)`` with shape ``(M + 1,)``. If None,
 52        initializes with zeros.
 53
 54    Notes
 55    -----
 56    At iteration ``k``, form the projection matrix and desired vector:
 57
 58    - ``X(k) ∈ C^{(L+1) x (M+1)}``, whose rows are regressor vectors, with the most
 59      recent regressor at row 0.
 60    - ``d_vec(k) ∈ C^{L+1}``, stacking the most recent desired samples, with
 61      ``d[k]`` at index 0.
 62
 63    The projection output and error vectors are:
 64
 65    .. math::
 66        y_{vec}(k) = X(k)\\,w^*(k) \\in \\mathbb{C}^{L+1},
 67
 68    .. math::
 69        e_{vec}(k) = d_{vec}(k) - y_{vec}(k).
 70
 71    The update direction ``u(k)`` is obtained by solving the regularized system:
 72
 73    .. math::
 74        (X(k)X^H(k) + \\gamma I_{L+1})\\,u(k) = e_{vec}(k),
 75
 76    and the coefficient update is:
 77
 78    .. math::
 79        w(k+1) = w(k) + \\mu X^H(k)\\,u(k).
 80
 81    This implementation returns only the *most recent* scalar components:
 82
 83    - ``y[k] = y_vec(k)[0]``
 84    - ``e[k] = e_vec(k)[0]``
 85
 86    References
 87    ----------
 88    .. [1] P. S. R. Diniz, *Adaptive Filtering: Algorithms and Practical
 89       Implementation*, 5th ed., Algorithm 4.6.
 90    """
 91
 92    supports_complex: bool = True
 93
 94    step_size: float
 95    gamma: float
 96    memory_length: int
 97
 98    def __init__(
 99        self,
100        filter_order: int,
101        step_size: float = 1e-2,
102        gamma: float = 1e-6,
103        L: int = 2,
104        w_init: Optional[ArrayLike] = None,
105    ) -> None:
106        super().__init__(filter_order=int(filter_order), w_init=w_init)
107        self.step_size = float(step_size)
108        self.gamma = float(gamma)
109        self.memory_length = int(L)
110
111    @validate_input
112    def optimize(
113        self,
114        input_signal: np.ndarray,
115        desired_signal: np.ndarray,
116        verbose: bool = False,
117        return_internal_states: bool = False,
118    ) -> OptimizationResult:
119        """
120        Executes the Affine Projection adaptation loop over paired input/desired sequences.
121
122        Parameters
123        ----------
124        input_signal : array_like of complex
125            Input sequence ``x[k]`` with shape ``(N,)`` (will be flattened).
126        desired_signal : array_like of complex
127            Desired sequence ``d[k]`` with shape ``(N,)`` (will be flattened).
128        verbose : bool, optional
129            If True, prints the total runtime after completion.
130        return_internal_states : bool, optional
131            If True, includes the last internal states in ``result.extra``:
132            ``"last_regressor_matrix"`` (``X(k)``) and
133            ``"last_correlation_matrix"`` (``X(k)X^H(k) + gamma I``).
134
135        Returns
136        -------
137        OptimizationResult
138            Result object with fields:
139            - outputs : ndarray of complex, shape ``(N,)``
140                Scalar output sequence, ``y[k] = y_vec(k)[0]``.
141            - errors : ndarray of complex, shape ``(N,)``
142                Scalar a priori error sequence, ``e[k] = e_vec(k)[0]``.
143            - coefficients : ndarray of complex
144                Coefficient history recorded by the base class.
145            - error_type : str
146                Set to ``"a_priori"``.
147            - extra : dict, optional
148                Present only if ``return_internal_states=True``.
149        """
150        tic: float = perf_counter()
151
152        dtype = complex
153        x = np.asarray(input_signal, dtype=dtype).ravel()
154        d = np.asarray(desired_signal, dtype=dtype).ravel()
155        
156        n_samples: int = int(x.size)
157        m: int = int(self.filter_order)
158        L: int = int(self.memory_length)
159
160        outputs: np.ndarray = np.zeros(n_samples, dtype=dtype)
161        errors: np.ndarray = np.zeros(n_samples, dtype=dtype)
162
163        x_padded: np.ndarray = np.zeros(n_samples + m, dtype=dtype)
164        x_padded[m:] = x
165
166        X_matrix: np.ndarray = np.zeros((L + 1, m + 1), dtype=dtype)
167        D_vector: np.ndarray = np.zeros(L + 1, dtype=dtype)
168
169        last_corr: Optional[np.ndarray] = None
170
171        eye_L: np.ndarray = np.eye(L + 1, dtype=dtype)
172
173        for k in range(n_samples):
174            X_matrix[1:] = X_matrix[:-1]
175            X_matrix[0] = x_padded[k : k + m + 1][::-1]
176
177            D_vector[1:] = D_vector[:-1]
178            D_vector[0] = d[k]
179
180            Y_vector: np.ndarray = X_matrix @ self.w.conj()
181            E_vector: np.ndarray = D_vector - Y_vector
182
183            outputs[k] = Y_vector[0]
184            errors[k] = E_vector[0]
185
186            corr_matrix: np.ndarray = (X_matrix @ X_matrix.conj().T) + (self.gamma * eye_L)
187            last_corr = corr_matrix
188
189            try:
190                u: np.ndarray = np.linalg.solve(corr_matrix, E_vector)
191            except np.linalg.LinAlgError:
192                u = np.linalg.pinv(corr_matrix) @ E_vector
193
194            self.w = self.w + self.step_size * (X_matrix.conj().T @ u)
195            self._record_history()
196
197        runtime_s: float = perf_counter() - tic
198        if verbose:
199            print(f"[AffineProjection] Completed in {runtime_s * 1000:.02f} ms")
200
201        extra = None
202        if return_internal_states:
203            extra = {
204                "last_regressor_matrix": X_matrix.copy(),
205                "last_correlation_matrix": None if last_corr is None else last_corr.copy(),
206            }
207
208        return self._pack_results(
209            outputs=outputs,
210            errors=errors,
211            runtime_s=runtime_s,
212            error_type="a_priori",
213            extra=extra,
214        )

 28class SignData(AdaptiveFilter):
 29    """
 30    Complex Sign-Data LMS adaptive filter.
 31
 32    Low-complexity LMS variant in which the regressor vector is replaced by its
 33    element-wise sign. This reduces multiplications (since the update uses a
 34    ternary/sign regressor), at the expense of slower convergence and/or larger
 35    steady-state misadjustment in many scenarios.
 36
 37    Parameters
 38    ----------
 39    filter_order : int
 40        Adaptive FIR filter order ``M``. The number of coefficients is ``M + 1``.
 41    step_size : float, optional
 42        Adaptation step size ``mu``. Default is 1e-2.
 43    w_init : array_like of complex, optional
 44        Initial coefficient vector ``w(0)`` with shape ``(M + 1,)``. If None,
 45        initializes with zeros.
 46
 47    Notes
 48    -----
 49    At iteration ``k``, form the regressor vector (newest sample first):
 50
 51    .. math::
 52        x_k = [x[k], x[k-1], \\ldots, x[k-M]]^T \\in \\mathbb{C}^{M+1}.
 53
 54    The a priori output and error are
 55
 56    .. math::
 57        y[k] = w^H[k] x_k, \\qquad e[k] = d[k] - y[k].
 58
 59    Define the element-wise sign regressor ``\\operatorname{sign}(x_k)``.
 60    The update implemented here is
 61
 62    .. math::
 63        w[k+1] = w[k] + 2\\mu\\, e^*[k] \\, \\operatorname{sign}(x_k).
 64
 65    Implementation details
 66        - For complex inputs, ``numpy.sign`` applies element-wise and returns
 67          ``x/|x|`` when ``x != 0`` and ``0`` when ``x == 0``.
 68        - The factor ``2`` in the update matches the implementation in this
 69          module (consistent with common LMS gradient conventions).
 70
 71    References
 72    ----------
 73    .. [1] P. S. R. Diniz, *Adaptive Filtering: Algorithms and Practical
 74       Implementation*, 5th ed., Algorithm 4.1 (sign-based LMS variants).
 75    """
 76
 77    supports_complex: bool = True
 78
 79    def __init__(
 80        self,
 81        filter_order: int,
 82        step_size: float = 1e-2,
 83        w_init: Optional[ArrayLike] = None,
 84    ) -> None:
 85        super().__init__(filter_order=int(filter_order), w_init=w_init)
 86        self.step_size = float(step_size)
 87
 88    @validate_input
 89    def optimize(
 90        self,
 91        input_signal: np.ndarray,
 92        desired_signal: np.ndarray,
 93        verbose: bool = False,
 94        return_internal_states: bool = False,
 95    ) -> OptimizationResult:
 96        """
 97        Executes the Sign-Data LMS adaptation loop over paired input/desired sequences.
 98
 99        Parameters
100        ----------
101        input_signal : array_like of complex
102            Input sequence ``x[k]`` with shape ``(N,)`` (will be flattened).
103        desired_signal : array_like of complex
104            Desired sequence ``d[k]`` with shape ``(N,)`` (will be flattened).
105        verbose : bool, optional
106            If True, prints the total runtime after completion.
107        return_internal_states : bool, optional
108            If True, includes the last internal state in ``result.extra``:
109            ``"last_sign_regressor"`` (``sign(x_k)``).
110
111        Returns
112        -------
113        OptimizationResult
114            Result object with fields:
115            - outputs : ndarray of complex, shape ``(N,)``
116                Scalar output sequence, ``y[k] = w^H[k] x_k``.
117            - errors : ndarray of complex, shape ``(N,)``
118                Scalar a priori error sequence, ``e[k] = d[k] - y[k]``.
119            - coefficients : ndarray of complex
120                Coefficient history recorded by the base class.
121            - error_type : str
122                Set to ``"a_priori"``.
123            - extra : dict, optional
124                Present only if ``return_internal_states=True``.
125        """
126        t0 = perf_counter()
127
128        x = np.asarray(input_signal, dtype=complex).ravel()
129        d = np.asarray(desired_signal, dtype=complex).ravel()
130
131        n_samples = int(x.size)
132        m = int(self.filter_order)
133
134        outputs = np.zeros(n_samples, dtype=complex)
135        errors = np.zeros(n_samples, dtype=complex)
136
137        x_padded = np.zeros(n_samples + m, dtype=complex)
138        x_padded[m:] = x
139
140        last_sign_xk: Optional[np.ndarray] = None
141
142        for k in range(n_samples):
143            x_k = x_padded[k : k + m + 1][::-1]
144
145            y_k = complex(np.vdot(self.w, x_k))
146            outputs[k] = y_k
147
148            e_k = d[k] - y_k
149            errors[k] = e_k
150
151            sign_xk = np.sign(x_k)
152            last_sign_xk = sign_xk
153
154            self.w = self.w + (2.0 * self.step_size) * np.conj(e_k) * sign_xk
155            self._record_history()
156
157        runtime_s = float(perf_counter() - t0)
158        if verbose:
159            print(f"[SignData] Completed in {runtime_s * 1000:.03f} ms")
160
161        extra: Optional[Dict[str, Any]] = None
162        if return_internal_states:
163            extra = {"last_sign_regressor": None if last_sign_xk is None else last_sign_xk.copy()}
164
165        return self._pack_results(
166            outputs=outputs,
167            errors=errors,
168            runtime_s=runtime_s,
169            error_type="a_priori",
170            extra=extra,
171        )